package com.flame.system.monitor.controller;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import com.flame.core.protocol.Result;
import com.flame.system.monitor.config.ZkConfigurationProperties;
import com.flame.system.monitor.controller.vo.ZookeeperNode;
import com.flame.system.monitor.exception.MonitorExceptionMessage;
import com.flame.system.monitor.exception.MonitorRuntimeException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.admin.ZooKeeperAdmin;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ZkManageController implements Watcher {

    private static final Logger log = LoggerFactory.getLogger(ZkManageController.class);

    private ZooKeeper zkClient;

    @Autowired
    private ZkConfigurationProperties zkConfigurationProperties;

    @PostConstruct
    public void init() throws IOException {
        if (zkConfigurationProperties.isEnabled()) {
            zkClient = new ZooKeeperAdmin(zkConfigurationProperties.getHost(), 60000, this);
        }
    }

    @PreDestroy
    public void close() throws InterruptedException {
        if (zkClient != null) {
            zkClient.close();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        if (Event.KeeperState.Disconnected.equals(event.getState()) ||
            Event.KeeperState.Expired.equals(event.getState())) {
            try {
                TimeUnit.SECONDS.sleep(3L);
                zkClient.close();
                zkClient = new ZooKeeperAdmin(zkConfigurationProperties.getHost(), 4000, this);
                log.info("zk reconnect success");
            } catch (IOException | InterruptedException e) {
                log.error("zk reconnect failed");
                e.printStackTrace();
            }
        }
    }

    @PostMapping("/system/zookeeper/node/tree")
    public Result nodeTree() throws InterruptedException, KeeperException {
        return Result.success(getTree("/"));
    }

    @PostMapping("/system/zookeeper/node/children")
    public Result nodeChildren(@RequestBody ZookeeperNode node) throws InterruptedException, KeeperException {
        return Result.success(zkClient.getChildren(node.getPath(), false));
    }

    private ZookeeperNode getTree(String path) throws InterruptedException, KeeperException {
        ZookeeperNode zookeeperNode = new ZookeeperNode();
        zookeeperNode.setPath(path);
        List<String> children = zkClient.getChildren(path, false);
        if (!CollectionUtils.isEmpty(children)) {
            List<ZookeeperNode> nodeList = new ArrayList<>();
            for (String child : children) {
                if ("/".equals(path)) {
                    ZookeeperNode node = getTree(path + child);
                    nodeList.add(node);
                } else {
                    ZookeeperNode node = getTree(path + "/" + child);
                    nodeList.add(node);
                }
            }
            zookeeperNode.setChildren(nodeList);
        }
        return zookeeperNode;
    }

    @PostMapping("/system/zookeeper/node")
    public Result getNode(@RequestBody ZookeeperNode node) throws InterruptedException, KeeperException {
        Stat nodeStat = zkClient.exists(node.getPath(), false);
        if (nodeStat == null) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.NODE_NOT_EXIST);
        }
        byte[] data = zkClient.getData(node.getPath(), false, new Stat());
        if (data != null) {
            node.setData(new String(data, StandardCharsets.UTF_8));
        }
        return Result.success(node);
    }

    @PostMapping("/system/zookeeper/node/create")
    public Result createNode(@RequestBody ZookeeperNode node) throws InterruptedException, KeeperException {
        if (!node.getPath().startsWith("/")) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.PATH_INVALID);
        }
        Stat nodeStat = zkClient.exists(node.getPath(), false);
        if (nodeStat != null) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.NODE_EXIST);
        }
        zkClient.create(node.getPath(), node.getData().getBytes(StandardCharsets.UTF_8), ZooDefs.Ids.OPEN_ACL_UNSAFE,
            CreateMode.PERSISTENT);
        return Result.success();
    }

    @PostMapping("/system/zookeeper/node/update")
    public Result updateNode(@RequestBody ZookeeperNode node) throws InterruptedException, KeeperException {
        Stat nodeStat = zkClient.exists(node.getPath(), false);
        if (nodeStat == null) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.NODE_NOT_EXIST);
        }
        zkClient.setData(node.getPath(), node.getData().getBytes(StandardCharsets.UTF_8), -1);
        return Result.success();
    }

    @PostMapping("/system/zookeeper/node/delete")
    public Result deleteNode(@RequestBody ZookeeperNode node) throws InterruptedException, KeeperException {
        Stat nodeStat = zkClient.exists(node.getPath(), false);
        if (nodeStat == null) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.NODE_NOT_EXIST);
        }
        List<String> children = zkClient.getChildren(node.getPath(), false);
        if (!CollectionUtils.isEmpty(children)) {
            throw new MonitorRuntimeException(MonitorExceptionMessage.CHILDREN_NOT_EMPTY);
        }
        zkClient.delete(node.getPath(), -1);
        return Result.success();
    }
}
